package week13.queue.memory;

import week13.queue.api.*;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

public class MemoryBroker implements Broker {


    private final Map<String, Storage> storageMap = new ConcurrentHashMap<>();
    private final Offset offset ;
    public MemoryBroker() {
        offset = new MemoryOffset();
    }

    @Override
    public Producer getProducer(String topicName) {
        return new MemoryProducer(getStorage(topicName));
    }

    @Override
    public Consumer getConsumer(String topicName,ConsumerGroup consumerGroup) {
        return new MemoryConsumer(getStorage(topicName),consumerGroup,offset);
    }

    private Storage getStorage(String topicName){
        return storageMap.computeIfAbsent(topicName, k -> new MemoryStorage());
    }
}
